package com.example.jobback.websocket.schedule;

import com.alibaba.fastjson.JSON;
import com.example.jobback.util.CACHEUtil;
import com.example.jobback.websocket.dto.MessageSendDto;
import com.example.jobback.websocket.mapper.MessageMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static com.example.jobback.websocket.util.RedisStreamConstants.*;

/**
 * @ClassName: UnreadMessageTask
 * @Description:
 * @Author: lhb
 * @Date: 2025/9/20
 */
@Component
@Slf4j
public class UnreadMessageTask {

    private final CACHEUtil cacheUtil;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    /**
     * @Description: 定时设置消息为已读的时间间隔
     * @Author lhb
     * @CreateTime 2025/9/21 0:48
     */
    private static final long PROCESS_UNREAD_MESSAGES = 3000;

    private static final long TRIM_STREAM = 60000;

    private final MessageMapper messageMapper;

    public UnreadMessageTask(CACHEUtil cacheUtil, MessageMapper messageMapper) {
        this.cacheUtil = cacheUtil;
        this.messageMapper = messageMapper;
    }


    /**
     * @Description: 定时设置消息为已读
     * @return void
     * @Author lhb
     * @CreateTime 2025/9/20 23:46
     */
    @Scheduled(fixedRate = PROCESS_UNREAD_MESSAGES)
    public void processUnreadMessages() {
        try {
            // 从Stream读取消息
            List<MapRecord<String, Object, Object>> records = cacheUtil.readStreamMessages(
                    UNREAD_MESSAGE_STREAM,
                    UNREAD_MESSAGE_CONSUMER_GROUP,
                    UNREAD_MESSAGE_CONSUMER,
                    BATCH_PROCESS_SIZE
            );
            if (records == null || records.isEmpty()) {
                return;
            }
            // 转换为对象列表
            List<MessageSendDto> messageList = records.stream()
                    .map(record -> {
                        MessageSendDto dto = new MessageSendDto();
                        // 设置四个核心字段
                        Map<Object, Object> values = record.getValue();
                        dto.setFromUserId(Long.parseLong((String) values.get("fromUserId")));
                        dto.setFromRole((String) values.get("fromRole"));
                        dto.setToUserId(Long.parseLong((String) values.get("toUserId")));
                        dto.setToRole((String) values.get("toRole"));
                        dto.setCreatedAt(LocalDateTime.parse(values.get("createdAt").toString()));
                        return dto;
                    })
                    .collect(Collectors.toList());
            // 批量更新消息为未读状态
            messageMapper.batchUpdateUnreadStatus(messageList);

            // 确认消息已处理
            String[] Ids = records.stream()
                    .map(record -> record.getId().toString())
                    .toArray(String[]::new);

            cacheUtil.acknowledgeStreamMessage(
                    UNREAD_MESSAGE_STREAM,
                    UNREAD_MESSAGE_CONSUMER_GROUP,
                    Ids
            );
            log.info("批量处理未读消息完成，处理数量: {}", records.size());
        } catch (Exception e) {
            log.error("处理未读消息失败", e);
        }
    }


    /**
     * @Description: 定时修剪 
     * @return void
     * @Author lhb
     * @CreateTime 2025/9/21 0:46
     */
    @Scheduled(fixedRate = TRIM_STREAM)
    public void trimStream() {
        try {
            // XTRIM命令：限制Stream长度，保留最新的maxLength条消息
            // approxTrim为true表示近似修剪（性能更好），false表示精确修剪
            Long currentLength = stringRedisTemplate.opsForStream().size(UNREAD_MESSAGE_STREAM);
            log.info("当前Stream长度: {}", currentLength);
            if(currentLength > STREAM_MAX_LENGTH) {
                Long trim = stringRedisTemplate.opsForStream().trim(UNREAD_MESSAGE_STREAM, STREAM_REMAIN_LENGTH, false);
                log.info("Stream修剪完成，保留最新的{}条消息，删除了{}条消息", STREAM_REMAIN_LENGTH, trim);
            }
        } catch (Exception e) {
            log.error("Stream修剪失败", e);
        }
    }



}
